package com.chc.sparrow;

import com.chc.domain.SensorReading;
import org.apache.flink.api.common.eventtime.BoundedOutOfOrdernessWatermarks;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

public class WatermarkApp {


    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());


        SingleOutputStreamOperator<Tuple2<String, Integer>> ds1 = env.fromElements(
                Tuple2.of("a", 1),
                Tuple2.of("a", 2),
                Tuple2.of("b", 3),
                Tuple2.of("c", 4)
        ).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<String, Integer>>forMonotonousTimestamps()
                .withTimestampAssigner((e, v) -> e.f1 * 1000L)
                //多并行度下水位线倾斜不能更新问题
                //在多并行度下，可能发生一个并行度没有数据，导致下游水位线更新卡主（一直在等待空闲任务发送数据）
                //必须设置这个参数，等待这个时间就忽略上游任务水位线了
                .withIdleness(Duration.ofSeconds(10)));

        SingleOutputStreamOperator<Tuple3<String, Integer, Integer>> ds2 = env.fromElements(
                Tuple3.of("a", 1, 1),
                Tuple3.of("a", 11, 1),
                Tuple3.of("b", 2, 1),
                Tuple3.of("b", 12, 1),
                Tuple3.of("c", 14, 1),
                Tuple3.of("d", 15, 1)
        ).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, Integer, Integer>>forMonotonousTimestamps()
                .withTimestampAssigner((e, v) -> e.f1 * 1000L)
                //多并行度下水位线倾斜不能更新问题
                //在多并行度下，可能发生一个并行度没有数据，导致下游水位线更新卡主（一直在等待空闲任务发送数据）
                //必须设置这个参数，等待这个时间就忽略上游任务水位线了
                .withIdleness(Duration.ofSeconds(10)));


    }
}
